package com.atguigu.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 */
public class MyPuntuatedWatermarkGenerator<T> implements WatermarkGenerator<T> {
  
  // 乱序等待时间
  private long delayTs;
  // 用来保存 当前为止 最大的事件时间
  private long maxTs;
  
  public MyPuntuatedWatermarkGenerator(long delayTs) {
    this.delayTs = delayTs;
    this.maxTs = Long.MIN_VALUE + this.delayTs + 1;
  }
  
  /**
   * 每条数据来，都会调用一次： 用来提取最大的事件时间，保存下来,并发射watermark
   *
   * @param event
   * @param eventTimestamp 提取到的数据的 事件时间
   * @param output
   */
  @Override
  public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
    maxTs = Math.max(maxTs, eventTimestamp);
    output.emitWatermark(new Watermark(maxTs - delayTs - 1));
    System.out.println("调用onEvent方法，获取目前为止的最大时间戳=" + maxTs + ",watermark=" + (maxTs - delayTs - 1));
  }
  
  /**
   * 周期性调用： 不需要
   *
   * @param output
   */
  @Override
  public void onPeriodicEmit(WatermarkOutput output) {
  
  }
}
